package pipeline;


import java.util.LinkedList;
import java.util.Queue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;

/**
 * Created by Administrator on 2017/3/17.
 */
public class SimplePipeline<T,OUT> extends AbstractPipe<T,OUT> implements Pipeline<T,OUT> {

    private final Queue<Pipe<?,?>> pipes = new LinkedList<Pipe<?, ?>>();

    private final ExecutorService helperExecutor;

    public SimplePipeline() {
        this(Executors.newSingleThreadExecutor(new ThreadFactory() {

            @Override
            public Thread newThread(Runnable r) {
                Thread t = new Thread(r,"SimplePipeline-Helper");
                t.setDaemon(true);
                return t;
            }

        }));
    }

    public SimplePipeline(final ExecutorService helperExecutor){
        super();
        this.helperExecutor = helperExecutor;
    }

    @Override
    public void shutdown(long timeout, TimeUnit timeUnit) {
        Pipe<?,?> pipe;
        while (null != (pipe = pipes.poll())){
            pipe.shutdown(timeout,timeUnit);
        }
        helperExecutor.shutdown();
    }

    @Override
    protected OUT doProcess(T input) throws PipeException {

        return null;
    }

    @Override
    public void addPipe(Pipe<?, ?> pipe) {
        pipes.add(pipe);
    }


    public <INPUT,OUTPUT> void addAsWorkerThreadBasePipe(Pipe<INPUT,OUTPUT> delegate ,int workerCount){
        addPipe(new WorkerThreadPipeDecorator<INPUT, OUTPUT>(delegate,workerCount));
    }

    public <INPUT,OUTPUT> void addAsThreadPoolBasedPipe(Pipe<INPUT,OUTPUT> delegate ,ExecutorService executorService){
        addPipe(new ThreadPoolPipeDecorator<INPUT, OUTPUT>(delegate,executorService));
    }

    @Override
    public void process(T input) throws InterruptedException {
        Pipe<T,?> firstPipe = (Pipe<T,?>) pipes.peek();
        firstPipe.process(input);
    }

    @Override
    public void init(final PipeContext pipeContext) {
        LinkedList<Pipe<?,?>> pipeList = (LinkedList<Pipe<?,?>>) pipes;

        Pipe<?,?> prevPipe = this ;
        for (Pipe<?,?> pipe : pipeList){
            prevPipe.setNextPipe(pipe);
            prevPipe=pipe;
        }

        Runnable task = new Runnable() {
            @Override
            public void run() {
                for (Pipe<?,?> pipe : pipes){
                    pipe.init(pipeContext);
                }
            }
        };

        helperExecutor.submit(task);

    }

    public PipeContext newDefaultPipeContext(){
        return new PipeContext() {
            @Override
            public void handleError(final PipeException exception) {
                helperExecutor.submit(new Runnable() {
                    @Override
                    public void run() {
                        exception.printStackTrace();
                    }
                });
            }
        };
    }
}
